/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dal_cyclonedds/utils/ParticipantMonitor.hpp"
#include "dal_cyclonedds/utils/ParticipantGuidMap.hpp"

ParticipantMonitor::~ParticipantMonitor()
{
    callback_info_s* p = nullptr;
    for (int i = 0; i < this->callback_info_list_.size(); i++) {
        p = this->callback_info_list_[i];
        delete p;
        p = nullptr;
        this->callback_info_list_[i] = nullptr;
    }
    this->callback_info_list_.clear();
}

ParticipantMonitor& ParticipantMonitor::GetInstance()
{
    static ParticipantMonitor singleton_;
    return singleton_;
}

void ParticipantMonitor::SetMonitor(dds_entity_t monitor_participant,
                                    dal_participant_monitor_callback callback,
                                    void* data)
{
    callback_info_s* callback_info = new callback_info_s;
    *callback_info = {callback, data, dds_time()};
    this->callback_info_list_.emplace_back(callback_info);
    dds_listener_t* c_listener = dds_create_listener(NULL);
    dds_lset_data_available_arg(
        c_listener, ParticipantMonitor::ProcessCallback, callback_info, true);
    dds_entity_t builtin_reader =
        dds_create_reader(monitor_participant, DDS_BUILTIN_TOPIC_DCPSPARTICIPANT, NULL, c_listener);
    if (builtin_reader < 0) {
        std::cout << "Initialize builtin reader failed." << std::endl;
    }
    return;
}

void ParticipantMonitor::ProcessCallback(dds_entity_t reader, void* arg)
{
    void* builtin_samples = nullptr;
    dds_sample_info_t builtin_infos;
    dds_builtintopic_participant_t* builtin_msg = nullptr;

    callback_info_s* callback_info = nullptr;
    int builtin_rc = 0;
    builtin_rc = dds_take(reader, &builtin_samples, &builtin_infos, 1, 1);
    if (builtin_rc > 0 &&
        (builtin_infos.instance_state == DDS_IST_NOT_ALIVE_DISPOSED || builtin_infos.valid_data)) {
        builtin_msg = (dds_builtintopic_participant_t*)builtin_samples;
        callback_info = reinterpret_cast<callback_info_s*>(arg);
    }
    if (callback_info == nullptr) {
        std::cout << "Builtin reader read nothing." << std::endl;
        return;
    }

    ParticipantMonitor& pm = ParticipantMonitor::GetInstance();
    std::lock_guard<std::mutex> lkg(pm.mutex_);
    dal_participant_monitor_info_t d_info;

    switch (builtin_infos.instance_state) {
        case DDS_IST_ALIVE:
            d_info.status = DAL_PARTICIPANT_ONLINE;
            break;
        default:
            d_info.status = DAL_PARTICIPANT_OFFLINE_ABNORMALLY;
            break;
    }

    d_info.participant_name.str = NULL;
    d_info.participant_name.length = 0;

    participant_born_info_s* part_info = nullptr;
    size_t part_info_len = sizeof(participant_born_info_s);
    dds_qget_userdata(builtin_msg->qos, (void**)&part_info, &part_info_len);
    memcpy((uint8_t*)&d_info.guid, builtin_msg->key.v, GUID_SIZE);

    // if dal_cyclonedds receive other DDS's pdp msg, just ignore it.
    if (!part_info) {
        dds_return_loan(reader, &builtin_samples, 1);
        return;
    }

    if (d_info.status == DAL_PARTICIPANT_ONLINE) {
        if (part_info->born_time_ < callback_info->reg_time_) {
            dds_free(part_info);
            dds_return_loan(reader, &builtin_samples, 1);
            return;
        }
        char* entity_name;
        dds_qget_entity_name(builtin_msg->qos, &entity_name);

        d_info.participant_name.str = entity_name;
        d_info.participant_name.length = strlen(entity_name);
        ParticipantGuidMapInsert(DalGuidForMap(d_info.guid), std::string(entity_name));
    } else if (d_info.status == DAL_PARTICIPANT_OFFLINE_ABNORMALLY) {
        std::string name;
        bool ret = ParticipantGuidMapFind(DalGuidForMap(d_info.guid), name);
        if (ret && name.length() > 0) {
            d_info.participant_name.str = (char *)malloc(name.length() + 1);
            memcpy(d_info.participant_name.str, name.c_str(), name.length());
            d_info.participant_name.length = name.length();
            d_info.participant_name.str[name.length()] = '\0';
        } else {
            d_info.participant_name.str = NULL;
            d_info.participant_name.length = 0;
        }
    }

    (callback_info->callback_)(&d_info, callback_info->data_);

    if (d_info.participant_name.str && d_info.participant_name.length > 0) {
        free(d_info.participant_name.str);
    }

    dds_return_loan(reader, &builtin_samples, 1);
    dds_free(part_info);

    return;
}
